package cn.codenest.util.kafka.impl;

import cn.codenest.db.DemoDB;
import cn.codenest.listener.SaveOffsetsOnReblance;
import cn.codenest.util.kafka.KafkaConsumerService;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author ：Hyman
 * @date ：Created in 2021/6/30 17:37
 * @description：
 * @modified By：
 * @version: $
 */
public class KafkaConsumerServiceImpl implements KafkaConsumerService {

    Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>(128);

    public <V> KafkaConsumer createConsumer() {
        Properties properties = new Properties();
        /* kafka的链接信息*/
        properties.put("bootstrap.servers", "172.17.6.85:9092");
        /* 消费者的群组信息，在同一个组的消费者会被分配到同一个主题的不同分区中 */
        properties.put("group.id", "myGroup");
        /* 指定消费者从服务器获取记录的最小字节数，代理在收到消费者的数据请求时，如果可用
         * 的数据量小于改值，则会等到有足够的可用数据时才会把他返回给消费者 */
        properties.put("fetch.min.bytes", "100");
        /* 如果没有足够的数据流入，消费者获取最小数据量得不到满足时，超过设置的fetch.max.wait.ms
         * 后，代理也会把现有的消息返给消费者 */
        properties.put("fetch.max.wait.ms", "500");
        /* 消费者给从每个分区获取的消息分配的最大内存，默认是1m，这个值要比max.message.size属性配置大，
         * 否则消费者可能无法读取这些消息，而一直在重试。*/
        properties.put("max.partition.fetch.bytes", "1048576");
        /* 消费者接受到的每个消息最大大小 */
        properties.put("max.message.size", "1024");
        /* 消费者在被认为死亡之前可以与服务器断开连接的时间，如果消费者没有在配置的时间内发送心跳
         * 给群组协调器，就被认为已经死亡，协调器就会触发再均衡。*/
        properties.put("session.timeout.ms", "30000");
        /* 指定了poll（）方法向协调器发送心跳的频率，一般要比session.timeout.ms要小，是其三分之一*/
        properties.put("heartbeat.interval.ms", "1000");
        /* 指定了消费者在读取一个没有偏移量的分区或者偏移量无效（因消费者长时间无效，包含偏移量的记录已经过时并被删除）
           的情况下应该如何处理，默认值是latest，在偏移量无效时消费者从最新的记录开始读；还可以设置earlist，意思是说，
           从起始位置开始读取分区记录 */
        properties.put("auto.offset.reset", "latest");
        /* 是否自动提交偏移量，默认是true；为了避免数据重复和丢失，可以设为false，由代码自己控制何时
         * 提交偏移量 */
        properties.put("enable.auto.commit", "true");
        /*自动提交偏移量的频率*/
        properties.put("auto.commit.interval.ms", "100");
        /* 将主题的分区分配给消费者的策略：
         * org.apache.kafka.clients.consumer.RangeAssignor：默认策略，假设2个主题,每个主题有3个分区，2个消费者订阅，则分区0和分区1会分配给消费者1，分区2会
         * 分配给消费者2.最后消费者1订阅了4个分区，消费者2订阅了2个分区。
         * org.apache.kafka.clients.consumer.RoundRobin：上述情况，消费者1会订阅主题1的分区0、分区2和主题2的分区1,消费者2会订阅主题1的分区1、主题2的
         * 分区0和分区2*/
        properties.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor");
        /* 可以是任意字符串，代理用来标识从客户端发送过来的消息，在日志、度量指标中使用 */
        properties.put("client.id", "consumer-client");
        /* 控制单词调用call()方法能够返回的记录数量，可以帮助控制在轮询中需要处理的数据量 */
        properties.put("max.poll.records", "10");
        /* 指定TCP Socket发送缓冲区的大小，被设置成-1后，使用操作系统的默认值
         *  receive.buffer.bytes是指定接收缓冲区的大小，如果生产者和消费者处于
         * 不同的数据中心，可以适当增大这些值，因为跨数据中心网络一般都有比较高的
         * 延迟和比较低的带宽*/
        properties.put("receive.buffer.bytes", "256");
        /* 偏移量自动提交的时间间隔默认是5s，每5秒自动提交偏移量 */
        properties.put("auto.commit.interval.ms", "5000");
        //properties.put("enable.auto.commit", "true");
        /* 关闭自动提交偏移量 */
        properties.put("enable.auto.commit", "false");
        /* 键反序列化器 */
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        /* 值反序列化器 */
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, V> consumer = new KafkaConsumer<String, V>(properties);
        return consumer;
    }

    @Override
    public void handle(String topic) {
        KafkaConsumer consumer1 = createConsumerAndSubcribe(topic);
        Thread t1 = new Thread(new MessageHandler(topic, consumer1, currentOffsets));

        KafkaConsumer consumer2 = createConsumerAndSubcribe(topic);
        Thread t2 = new Thread(new MessageHandler(topic, consumer2, currentOffsets));

        KafkaConsumer consumer3 = createConsumerAndSubcribe(topic);
        Thread t3 = new Thread(new MessageHandler(topic, consumer3, currentOffsets));
        t1.start();
        t2.start();
        t3.start();
        try {
            t1.join();
            t2.join();
            t3.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private KafkaConsumer createConsumerAndSubcribe(String topic) {
        KafkaConsumer consumer = createConsumer();
        /* 普通的再均衡监听器 */
        //ConsumerRebalanceListener consumerRebalanceListener = new HandleRebalanceListener(currentOffsets, consumer);
        //consumer.subscribe(Collections.singletonList(topic), consumerRebalanceListener);
        /* 可以自定义偏移量的再均衡监听器 */
        SaveOffsetsOnReblance saveOffsetsOnReblance = new SaveOffsetsOnReblance(currentOffsets, consumer);
        consumer.subscribe(Collections.singletonList(topic), saveOffsetsOnReblance);
        /* poll方法可以让消费者马上加入到群组中 */
        consumer.poll(0);
        /* 然后需要调用seek定位每个分区的偏移量 */
        for (Object obj : consumer.assignment()) {
            TopicPartition partition = (TopicPartition) obj;
            consumer.seek(partition, saveOffsetsOnReblance.getOffsetFromDB(partition));
        }
        return consumer;
    }

    public static class MessageHandler implements Runnable {

        String topic;
        KafkaConsumer consumer;

        private Map<TopicPartition, OffsetAndMetadata> currentOffsets;
        int count = 0;

        public MessageHandler(String topic, KafkaConsumer consumer, Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
            this.topic = topic;
            this.consumer = consumer;
            this.currentOffsets = currentOffsets;
        }

        @Override
        public void run() {
            try {
                handlingAndCommitOffsetInDB();
                //handledThenCommitAsync();
                //handlingAndCommitAsync();
                return;
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    /* 最后在关闭消费者前调用同步提交，确保关闭前会进行一次提交
                     * 如果提交失败，同步提交会不断重试，直到成功 */
                    consumer.commitSync();
                } finally {
                    consumer.close();
                }
            }
        }

        private void handlingAndCommitOffsetInDB() {
            while (true) {
                //poll方法里面的参数是超时时间，超过100毫秒，没有接受到数据，则返回
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                    storeRecordInDB(record);
                    storeOffsetInDB(record.topic(), record.partition(), record.offset());
                }
                commitDBTransaction();
            }
        }

        public void commitDBTransaction() {
            DemoDB.commitDBTransaction();
        }

        private void storeRecordInDB(ConsumerRecord<String, String> record) {
            DemoDB.consumerRecords.put(DemoDB.getKey(record.topic(), record.partition()), record);
        }

        private void storeOffsetInDB(String topic, int partition, long offset) {
            DemoDB.offsets.put(DemoDB.getKey(topic, partition), offset);
        }

        private void processRecord(ConsumerRecord<String, String> record) {
            System.out.println(
                    String.format("topic=%s,partition=%s,offset=%d,customer=%s,log=%s\n",
                            record.topic(),
                            record.partition(),
                            record.offset(),
                            record.key(),
                            record.value()));
        }

        /**
         * create by: Hyman
         * description: 单条处理完成之后，每2条偏移进行提交
         * create time: 2021/7/8
         */
        private void handlingAndCommitAsync() {
            while (true) {
                //poll方法里面的参数是超时时间，超过100毫秒，没有接受到数据，则返回
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                    /* 提交明确指定的偏移量，每2次提交一次 */
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1, "no metadata"));
                    if (count % 2 == 0) {
                        consumer.commitAsync(currentOffsets, null);
                    }
                    count++;
                }
                /* 为了确保消息处理完时必定提交，这里再异步提交一次。*/
                consumer.commitAsync();
            }
        }

        /**
         * create by: Hyman
         * description: 处理完成所有消息之后，统一进行异步提交
         * create time: 2021/7/8
         */
        private void handledThenCommitAsync() {
            while (true) {
                //poll方法里面的参数是超时时间，超过100毫秒，没有接受到数据，则返回
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }
                /*这是异步提交偏移量的方式，可以接受一个回调函数，提交完成之后调用该回调函数
                 * 异步提交没有重试机制，即使提交失败，也只会提交一次 */
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        if (e != null) {
                            System.out.println(String.format("提交错误:{}{}", map, e));
                        }
                    }
                });
            }
        }
    }
}
